package com.jcj.jcjmq.client;

import com.jcj.jcjmq.model.Message;
import lombok.Getter;

import java.util.concurrent.atomic.AtomicInteger;

public class JcjConsumer<T> {
    String id;
    JcjBroker broker;
    //    String topic;
    JcjMq mq;


    AtomicInteger index = new AtomicInteger(0);

    public JcjConsumer(JcjBroker broker) {
        this.broker = broker;
        this.id = "consumer-" + index.getAndIncrement();
    }

    public void sub(String topic) {
//        this.topic = topic;
        broker.sub(topic, id);
//        this.topic = topic;
//        mq = broker.find(topic);
//        if (mq == null) {
//            throw new RuntimeException("topic not found");
//        }
    }

    public Message<T> recv(String topic) {
        return broker.recv(topic, id);
//        return mq.poll(timeout);
    }

    @Getter
    JcjListener listener;

    //todo
    public void listener(String topic, JcjListener<T> listener) {
        this.listener = listener;
        broker.addConsumer(topic, this);
    }

    public void unsub(String topic) {
        broker.unsub(topic, id);
    }


    public boolean ack(String topic, int offset) {
        return broker.ack(topic, id, offset);
    }

    public boolean ack(String topic, Message message) {

        if (message == null || message.getHeaders() == null || message.getHeaders().get("x-offset") == null) {
            return false;
        }
        int offset = Integer.parseInt((String) message.getHeaders().get("x-offset"));
        return ack(topic, offset);
    }
}
